package tsif.tcluster.service.proxy.impl;

import tsif.tcluster.ITClusterAccessor;
import tsif.tcluster.TClusterContext;
import tsif.tcluster.rpc.TClusterRpcCallBackMsg;
import tsif.tcluster.rpc.TClusterRpcCallMsg;
import tsif.tcluster.rpc.TClusterRpcProtocol;
import tsif.tcluster.rpc.TClusterRpcUtils;
import tsif.tcluster.service.proxy.impl.TClusterCallbacks.CallBack;
import tsif.tcluster.session.TClusterSession;

/**
 * 集群环境
 */
public class TClusterAccessor implements ITClusterAccessor {
	protected TClusterRpcServiceProxy serviceProxy;
	protected int callTimeOut; // 阻塞访问超时时间

	public TClusterAccessor(TClusterRpcServiceProxy serviceProxy) {
		this.serviceProxy = serviceProxy;
		callTimeOut = 60 * 1000; // 默认超时时间
	}

	/** 获取连接所依赖的环境(本机的) **/
	public TClusterContext getContext() {
		return serviceProxy.getContext();
	}

	@Override
	public int getCallTimeOut() {
		return (callTimeOut > 1000) ? callTimeOut : 1000;
	}

	@Override
	public void setCallTimeOut(int callTimeOut) {
		this.callTimeOut = callTimeOut;
	}

	@Override
	public long newService(Class<?> classValue, Object... args) throws Exception {
		return getContext().newService(classValue, args);
	}

	@Override
	public long newService(long serviceId, Class<?> classValue, Object... args) throws Exception {
		return getContext().newService(serviceId, classValue, args);
	}

	@Override
	public long getUniqueServiceId(Object key) {
		return getContext().getUniqueServiceId(key);
	}

	@Override
	public long uniqueService(Object key, Class<?> classValue, Object... args) throws Exception {
		return getContext().uniqueService(key, classValue, args);
	}

	@Override
	public long uniqueService(long newServiceId, Object key, boolean replace, Class<?> classValue, Object... args) throws Exception {
		return getContext().uniqueService(newServiceId, key, replace, classValue, args);
	}

	@Override
	public <T> T call(Object service, String funcName, Object... args) {
		return call(service, getCallTimeOut(), funcName, args);
	}

	@SuppressWarnings("unchecked")
	@Override
	public <T> T call(Object service, final int timeout, String funcName, Object... args) {
		final SyncObject syncObj = new SyncObject();
		syncObj.finish = false;

		// send action
		send(service, funcName, new CallBack<T>() {
			@Override
			protected void onCallBack(int code, String msg, T retObj) {
				syncObj.code = code;
				syncObj.msg = msg;
				syncObj.obj = retObj;
				synchronized (syncObj) {
					syncObj.finish = true;
					syncObj.notify();
				}
			}

			@Override
			public int getMaxTimeOut() {
				return timeout;
			}

			@Override
			public boolean isDirectCall() {
				return true; // 阻塞callback, 不能放入线程池会阻塞.
			}
		}, args);

		// wait finish
		long startTime = System.currentTimeMillis();
		synchronized (syncObj) {
			while (!syncObj.finish) {
				try {
					syncObj.wait(timeout + 1000); // 这里加timeout是避免bug万一卡死了.
				} catch (InterruptedException e) {
					throw new RuntimeException(e);
				}
				// check finish again
				if (syncObj.finish) {
					break;
				}
				// 检测时间(最后容错, 万一线程被卡主会死掉)
				long nowTime = System.currentTimeMillis();
				long dt = nowTime - startTime;
				if (dt > timeout) {
					throw new RuntimeException("call no callback to time out(" + dt + "/" + timeout + ")! service=" + service + " funcName=" + funcName);
				}
			}
		}

		// check
		if (syncObj.code <= 0) {
			throw new RuntimeException("call error! code=" + syncObj.code + " msg=" + syncObj.msg);
		}
		return (T) syncObj.obj;
	}

	/** 阻塞调用, 使用Thread.yield()等待 **/
	@SuppressWarnings("unchecked")
	protected <T> T callByWait(Object service, final int timeout, String funcName, Object... args) {
		final SyncObject syncObj = new SyncObject();
		syncObj.finish = false;

		// send action
		send(service, funcName, new CallBack<T>() {
			@Override
			protected void onCallBack(int code, String msg, T retObj) {
				syncObj.code = code;
				syncObj.msg = msg;
				syncObj.obj = retObj;
				syncObj.finish = true;
			}

			@Override
			public int getMaxTimeOut() {
				return timeout;
			}

			@Override
			public boolean isDirectCall() {
				return true;
			}
		}, args);

		// wait finish
		while (!syncObj.finish) {
			try {
				Thread.yield();
			} catch (Exception e) {
				throw new RuntimeException(e);
			}
		}

		// check
		if (syncObj.code <= 0) {
			throw new RuntimeException("call error! code=" + syncObj.code + " msg=" + syncObj.msg);
		}
		return (T) syncObj.obj;
	}

	@Override
	public void send(Object service, String funcName, Object... args) {
		this.send(service, funcName, null, args);
	}

	@Override
	public <T> void send(Object service, String funcName, CallBack<T> callback, Object... args) {
		// add callback
		int callbackId = serviceProxy.addCallBack(callback);

		// rpc call msg
		TClusterRpcCallMsg rpcmsg = new TClusterRpcCallMsg();
		rpcmsg.setMethod(funcName);
		rpcmsg.setCallbackId(callbackId);
		rpcmsg.setArgDatas(args);

		// 判断数据是否要转成二进制数据
		Object data = rpcmsg;
		TClusterContext context = getContext();
		if (context.isUseServiceClassLoader() || !context.isLocalService(service)) {
			data = TClusterRpcUtils.toByte(rpcmsg, rpcmsg.getClass()); // 非本地服务或者本地使用不同的类加载器, 所以要转成二进制
		}
		// send packet
		context.send(serviceProxy.getServiceId(), service, TClusterRpcProtocol.RPC_CALL, data);

	}

	/** 发送回调消息 **/
	protected void callback(TClusterSession session, long serviceId, int callbackId, int code, String msg, Object retObj) {

		// rpc callback msg
		TClusterRpcCallBackMsg rpcmsg = new TClusterRpcCallBackMsg();
		rpcmsg.setCallbackId(callbackId);
		rpcmsg.setCode(code);
		rpcmsg.setMsg(msg);
		rpcmsg.setRetDatas(retObj);
		// System.out.println("callback data: " + Arrays.toString(retdatas) + (RpcUtils.toAutoObject(retdatas)));

		// 判断数据是否要转成二进制数据
		Object data = rpcmsg;
		TClusterContext context = getContext();
		if (context.isUseServiceClassLoader() || !context.isLocalService(serviceId)) {
			data = TClusterRpcUtils.toByte(rpcmsg, rpcmsg.getClass()); // 非本地服务或者本地使用不同的类加载器, 所以要转成二进制
		}
		// send packet
		session.send(serviceProxy.getServiceId(), serviceId, TClusterRpcProtocol.RPC_CALLBACK, data);
	}

	/** 异步对象, 阻塞返回 **/
	protected static class SyncObject {
		public volatile boolean finish; // 是否完成
		public int code; // 结果
		public String msg; // 错误消息
		public Object obj; // 正确对象
	}

}
